package com.atguigu.flink.sql.query;

import com.atguigu.flink.func.ClickSource;
import com.atguigu.flink.pojo.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * Created by 黄凯 on 2023/6/27 0027 18:15
 *
 * @author 黄凯
 * 永远相信美好的事情总会发生.
 * <p>
 * TopN语法:
 * *    特殊的over()语法.
 * *    正常情况下， over中的order by只能使用时间字段，且必须为asc。
 * *    如果基于row_number() 进行over()操作， 且会按照row_number的结果进行where( where rk <=n)过滤提取，
 * *    Flink就能识别该操作为TopN操作，此时， order by 后面可以使用其他字段，且可以使用desc 。
 */
public class Flink07_TopN {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStreamSource<Event> ds = env.addSource(new ClickSource());

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        Schema schema =
                Schema.newBuilder()
                        .column("user", "string")
                        .column("url", "string")
                        .column("ts", "bigint")
                        .columnByExpression("pt", "proctime()")
                        .columnByExpression("et", "to_timestamp_ltz(ts, 3)")
                        .watermark("et", "et - interval '0' second")
                        .build();
        Table table = tableEnv.fromDataStream(ds, schema);
        table.printSchema();
        tableEnv.createTemporaryView("t1", table);

        //Url点击的TopN
        //1. 使用窗口求每个url的点击次数
        String countSql =
                "select url,\n" +
                        "       count(url) cnt,\n" +
                        "       window_start,\n" +
                        "       window_end\n" +
                        "from table(\n" +
                        "        tumble(table t1,descriptor(et), interval '10' second)\n" +
                        "    )\n" +
                        "group by window_start, window_end, url";

        Table t2 = tableEnv.sqlQuery(countSql);
//        t2.execute().print();
        tableEnv.createTemporaryView("t2", t2);

        //2. 基于求出来的点击次数排序求排名
        String rankSql =
                "select url,\n" +
                        "       cnt,\n" +
                        "       window_start,\n" +
                        "       window_end,\n" +
                        "       row_number() over (partition by window_start,window_end order by cnt desc ) rk\n" +
                        "from t2";

        Table t3 = tableEnv.sqlQuery(rankSql);

        //The window can only be ordered in ASCENDING mode.
        //OVER windows' ordering in stream mode must be defined on a time attribute.
        //不能用非时间字段排序，到下面就识别了
//        t3.execute().print();
        tableEnv.createTemporaryView("t3", t3);

        //3. 取topn
        String topNSql =
                "select url,\n" +
                        "       cnt,\n" +
                        "       window_start,\n" +
                        "       window_end,\n" +
                        "       rk\n" +
                        "from t3\n" +
                        "where rk <= 2";

        Table result = tableEnv.sqlQuery(topNSql);
//        result.execute().print();

        //组合
        String sql =
                "select url,\n" +
                        "       cnt,\n" +
                        "       window_start,\n" +
                        "       window_end,\n" +
                        "       rk\n" +
                        "from (select url,\n" +
                        "             cnt,\n" +
                        "             window_start,\n" +
                        "             window_end,\n" +
                        "             row_number() over (partition by window_start,window_end order by cnt desc ) rk\n" +
                        "      from (select url,\n" +
                        "                   count(url) cnt,\n" +
                        "                   window_start,\n" +
                        "                   window_end\n" +
                        "            from table(\n" +
                        "                    tumble(table t1,descriptor(et), interval '10' second)\n" +
                        "                )\n" +
                        "            group by window_start, window_end, url) t2) t3\n" +
                        "where rk <= 2";

        Table table1 = tableEnv.sqlQuery(sql);
        table1.execute().print();


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }

    }

}
